package com.lagou.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Iterator;

public class TumblingWindowCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> dataStreamSource = executionEnvironment.socketTextStream("hdp-1", 9999);
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        SingleOutputStreamOperator<Tuple2<String, String>> mapDataStream = dataStreamSource.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String s) throws Exception {
                String dateString = simpleDateFormat.format(System.currentTimeMillis());
                return new Tuple2<>(s, dateString);
            }
        });

        //转换数据
        KeyedStream<Tuple2<String, String>, String> keyedStream = mapDataStream.keyBy(value -> value.f0);

        //使用窗口参数，但是由于之前使用了keyBy，所以是同一个slot达到3个数据才输出
        WindowedStream<Tuple2<String, String>, String, GlobalWindow> windowedStream = keyedStream.countWindow(3);

        SingleOutputStreamOperator<String> apply = windowedStream.apply(new WindowFunction<Tuple2<String, String>, String, String, GlobalWindow>() {
            @Override
            public void apply(String s, GlobalWindow window, Iterable<Tuple2<String, String>> input, Collector<String> out) throws Exception {
                StringBuilder sb = new StringBuilder();
                Iterator<Tuple2<String, String>> iterator = input.iterator();
                while (iterator.hasNext()) {
                    Tuple2<String, String> value = iterator.next();
                    sb.append(value.f0 + "..." + value.f1);
                    sb.append("\t");
                }
                out.collect(sb.toString());
            }
        });

        apply.print();
        executionEnvironment.execute();

    }
}
